package jrain.iot.server.mqtt.service;

import java.util.HashMap;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import jrain.fw.core.business.BusinessConst;
import jrain.fw.core.log.Logger;
import jrain.fw.core.log.LoggerFactory;
import jrain.fw.core.utils.FwUtils;
import jrain.iot.server.mqtt.handler.MqttHandler;
import jrain.utils.collection.RDMap;
import jrain.utils.collection.RMap;
import jrain.utils.lang.LangUtils;

public class MqttNettyServer {
	private static final Logger logger = LoggerFactory.getLogger(MqttNettyServer.class);
	private volatile EventLoopGroup bossGroup;
	private volatile EventLoopGroup workerGroup;
	private volatile ChannelFuture f;

	public void start() {
		try {
			RMap<String, Object> cfgMap = FwUtils.getCfgService().getCfgMap("mqtt", new RDMap<>(new HashMap<>()));
			boolean enable = LangUtils.parseBoolean(cfgMap.get("mqtt.enable"), false);
			if (enable) {
				bossGroup = new NioEventLoopGroup();
				workerGroup = new NioEventLoopGroup();
				int port = LangUtils.parseInt(cfgMap.get("mqtt.server.port"), 6158);
				int idle = LangUtils.parseInt(cfgMap.get("mqtt.server.idle"), 60);
				ServerBootstrap b = new ServerBootstrap();
				b.group(bossGroup, workerGroup);
				b.channel(NioServerSocketChannel.class);
				b.childHandler(new ChannelInitializer<SocketChannel>() {
					@Override
					public void initChannel(SocketChannel ch) throws Exception {
						ch.pipeline().addLast("mqttEncoder", MqttEncoder.INSTANCE);
						ch.pipeline().addLast("mqttDecoder", new MqttDecoder());
						ch.pipeline().addLast("idleState", new IdleStateHandler(0, 0, idle));
						ch.pipeline().addLast("mqttHandler", new MqttHandler());
					}
				});
				b.option(ChannelOption.SO_BACKLOG, 128);
				b.childOption(ChannelOption.SO_KEEPALIVE, true);
				f = b.bind(port).sync();
				logger.info(BusinessConst.LOG_BOOT_REQUEST, "mqtt server started!,port={}", port);
				f.channel().closeFuture().sync();
			}
		} catch (Exception e) {
			e.printStackTrace();
			logger.error(BusinessConst.LOG_BOOT_REQUEST, "mqtt server start error!", e);
			stop();
		}
	}

	public void stop() {
		logger.info(BusinessConst.LOG_BOOT_REQUEST, "mqtt server stoped!");
		if (workerGroup != null) {
			workerGroup.shutdownGracefully();
		}
		if (bossGroup != null) {
			bossGroup.shutdownGracefully();
		}
		if (f != null) {
			f.channel().close();
		}
	}

	public static void main(String[] args) {
		new MqttNettyServer().start();
	}

}
